package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.ArticleConstants;
import com.heima.common.constants.BehaviorConstans;
import com.heima.model.message.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class KafkaStreamHelloListener {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //创建kstream对象，同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream(BehaviorConstans.HOT_ARTICLE_AGG_TOPIC);
        stream.map((key, value) -> {
                    Message message = new Message();
                    message = JSON.parseObject(value, Message.class);
                    return new KeyValue<>(String.valueOf(message.getArticleId()), JSON.toJSONString(message));
                })
                .groupBy((key, value) -> key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .aggregate(new Initializer<String>() {
                    @Override
                    public String apply() {
                        Message message = new Message();
                        return JSON.toJSONString(message);
                    }
                }, new Aggregator<String, String, String>() {
                    @Override
                    public String apply(String key, String message, String applyValue) {
                        log.info("开始聚合");
                        Message messageNew = JSON.parseObject(message, Message.class);
                        Message messageOld = JSON.parseObject(applyValue, Message.class);
                        messageOld.setCollect(messageOld.getCollect() + messageNew.getCollect());
                        messageOld.setComment(messageOld.getComment() + messageNew.getComment());
                        messageOld.setLike(messageOld.getLike() + messageNew.getLike());
                        messageOld.setView(messageOld.getView() + messageNew.getView());
                        messageOld.setArticleId(Long.valueOf(key));
                        return JSON.toJSONString(messageOld);
                    }
                }, Materialized.as("hot-atricle-stream-count-001"))
                .toStream()
                .map((key, value) -> {
                    log.info("聚合成功:{}", value);
                    return new KeyValue<>(key.key().toString(), value);
                })
                .to(ArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        return stream;
    }
}